package day05;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.AssignerWithPeriodicWatermarks;
import org.apache.flink.streaming.api.functions.AssignerWithPunctuatedWatermarks;
import org.apache.flink.streaming.api.functions.source.SourceFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.watermark.Watermark;
import org.apache.flink.streaming.api.windowing.time.Time;

import javax.annotation.Nullable;

public class TestWaterMark01 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(new Configuration());
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);//时间特性（语义） 默认 ProcessingTime
        env.getConfig().setAutoWatermarkInterval(1000);//间歇性生成Watermark的时间 默认200ms获取一次
        env.addSource(new SourceFunction<String>() {
            Boolean flag=true;
            int num = 0;
            @Override
            public void run(SourceContext<String> ctx) throws Exception {
             while (flag){
                 long processing = System.currentTimeMillis();
                 Long eventime=(processing-2000);//时间发生时间
                 Long water=eventime-2000;//认为时间发生2s前时间到了
                 String line=+num+"\thainiu\t"+eventime;
                 ctx.collectWithTimestamp(line,processing);//不仅能提交还能带时间戳
                // ctx.emitWatermark(new Watermark(water));

                 Thread.sleep(1000);
                 num++;
             }
            }

            @Override
            public void cancel() {
                flag=false;
            }
        })                                          //AssignerWithPunctuatedWatermarks   每过来一个拉一次
//                .assignTimestampsAndWatermarks(new AssignerWithPeriodicWatermarks<String>() {
//                                                                //间隔性的拉一次
//                                                                // 子类
//                                                                    //AscendingTimestampExtractor 升序的 watermark加与不加无所谓
//                                                                    //BoundedOutOfOrdernessTimestampExtractor 一定要加
//                    Long current_timestap=0L;
//                    @Nullable
//                    @Override//间歇性  生成Watermark 默认200ms获取一次
//                    public Watermark getCurrentWatermark() {
//                        //间隔时间获取waterMark'
//                        System.out.println("getCurrentWatermark");
//                        return new Watermark(current_timestap-2000);
//                    }
//
//                    @Override//提取时间戳 来一条提取一次(1s一条)
//                    public long extractTimestamp(String element, long previousElementTimestamp) {
//                        System.out.println("extractTimestamp");
//                        //数据获取时间戳
//                        String[] split = element.split("\t");
//                        Long current_timestap = Long.valueOf(split[2]);
//                        this.current_timestap= current_timestap;
//                        return 0;
//                    }
//                })
//                .print();
        /**
         * 确定性的
         */
//        .assignTimestampsAndWatermarks(new AssignerWithPunctuatedWatermarks<String>() {
//            @Nullable
//            @Override
//            public Watermark checkAndGetNextWatermark(String lastElement, long extractedTimestamp) {
//                return new Watermark(extractedTimestamp-2000);
//            }
//
//            @Override
//            public long extractTimestamp(String element, long previousElementTimestamp) {
//                System.out.println("extractTimestamp");
//                String[] split = element.split("\t");
//                Long current_timestap = Long.valueOf(split[2]);
//                return current_timestap;
//            }
//        }).print();


                /**
                 *
                 * 间歇性 升序的
                 */
//                .assignTimestampsAndWatermarks(new AscendingTimestampExtractor<String>() {
//                    @Override
//                    public long extractAscendingTimestamp(String element) {
//                        return Long.valueOf(element.split("\t")[2]);
//                        //watermark 为eventime - 1ms 认为（eventime - 1ms）及以前的人都来了
//                    }
//                }).print();

        .assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<String>(Time.seconds(2)) {
            @Override
            public long extractTimestamp(String element) {
                return Long.valueOf(element.split("\t")[2]);
            }
        }).print();



        env.execute();
    }
}
